package cn.oldsix.spark.streaming.flume.interception;

import cn.oldsix.spark.streaming.vo.UserBehaviour;
import com.alibaba.fastjson.JSON;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;

/*
 * @Author : Wu.D.J
 * @Description : 自定义用户行为拦截器
 * */

public class BehaviourInterceptor implements Interceptor {

    private Logger logger = LoggerFactory.getLogger(BehaviourInterceptor.class);

    @Override
    public void initialize() {

    }

    @Override
    public Event intercept(Event event) {

        if (event == null || event.getBody() == null || event.getBody().length == 0) {
            return null;
        }

        long userId = 0;

        String json = new String(event.getBody());
        String bodyJson = json.replaceAll("\\\\x22", "\"");
        event.setBody(bodyJson.getBytes());
        try {
            UserBehaviour behaviour = JSON.parseObject(bodyJson, UserBehaviour.class);
            userId = behaviour.getUserId();
        } catch (Exception e) {
            logger.error("event body is Invalid, bodyJson = [ " + bodyJson + "]", e);
        }

        if (userId == 0) {
            return null;
        }

        event.getHeaders().put("key", String.valueOf(userId));
        return event;
    }

    @Override
    public List<Event> intercept(List<Event> list) {
        List<Event> events = new ArrayList<>();
        for (Event event : list) {
            Event outEvent = intercept(event);
            if (outEvent != null) {
                events.add(outEvent);
            }
        }
        logger.info("event intercept success!");
        return events;
    }

    @Override
    public void close() {

    }

    public static class BehaviourInterceptorBuilder implements Builder {

        @Override
        public Interceptor build() {
            return new BehaviourInterceptor();
        }

        @Override
        public void configure(Context context) {

        }
    }

}
